package com.zy.datareceiver.binsess.stomp.service;

import com.zy.datareceiver.binsess.decode.service.StorageTaskService;
import com.zy.datareceiver.binsess.monitoring.service.RealMonitorService;
import com.zy.datareceiver.core.cast.ConCast;
import com.zy.datareceiver.core.common.ServerResponse;
import com.zy.datareceiver.core.common.SpringApplicationContext;
import com.zy.datareceiver.core.utils.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.SimpMessageSendingOperations;

import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Stomp数据推送线程，大屏幕和实时数据的推送线程核心
 * <p>其内部实现依靠于{@link org.springframework.messaging.simp.SimpMessageSendingOperations}
 *  从来源于{@link StorageTaskService#pointIdsQueue}队列中的数据取出新传过来数据的监测点pointId，
 *  根据pointId去组合数据，然后返回转换成Json字符串进行数据推送，
 *  大屏幕
 *      推送地址
 *          {@link ConCast.StompDestination#INDEX_LARGE_SCREEN_PATH},
 *      推送的主要数据包括
 *          实时数据{@link com.zy.datareceiver.binsess.monitoring.param.index.DetectionData}
 *          设备类型分布{@link com.zy.datareceiver.binsess.monitoring.param.index.EquipmentTypesDistribution}
 *          预警类型分布{@link com.zy.datareceiver.binsess.monitoring.param.index.WarningTypesDistributionData}
 *          预警信息统计{@link com.zy.datareceiver.binsess.monitoring.param.index.WarningStatisticsData}
 *      详细的数据实现，详见
 *      @See {@link com.zy.datareceiver.binsess.monitoring.service.impl.RealMonitorServiceImpl#getMonitoringIndexCollectData}
 *  实时数据监测
 *      推送地址
 *          {@link ConCast.StompDestination#TOPIC_PREFIX +"/"+ pointId},
 *      推送的主要数据包括
 *          实时数据{@link com.zy.datareceiver.binsess.monitoring.param.index.DetectionData}
 *          监测点详细信息{@link com.zy.datareceiver.binsess.monitoring.entity.MonitoringPoint}
 *          24小时预警数据{@link com.zy.datareceiver.binsess.warning.entity.WarningData}
 *      详细的数据实现，详见
 *      @See {@link com.zy.datareceiver.binsess.monitoring.service.impl.RealMonitorServiceImpl#getMonitoringCollectData(Integer)}
 *</p>
 */
public class StompMessageService extends Thread {
    @Resource
    private  SimpMessageSendingOperations messageTemplate  = SpringApplicationContext.getBean(SimpMessageSendingOperations.class);
    private  RealMonitorService realMonitorService = SpringApplicationContext.getBean(RealMonitorService.class);

    private Logger logger = LoggerFactory.getLogger(StompMessageService.class);
    private static volatile AtomicLong size = new AtomicLong(0);

    /**
     * @param topic 订阅的主题
     * @param info 需要推送的消息
     * @date 2018年11月6日
     * @throws NullPointerException 参数为null时
     */
    public void pushMessage(String topic,String info) {
        if (info == null || info.trim().length() <= 0) {
            return ;
        }
        if (ConCast.StompDestination.INDEX_LARGE_SCREEN_PATH.equals(topic)){
            size.incrementAndGet();
            logger.info("{} send message to {},message content {} ",size,topic, info);
        }
        messageTemplate.convertAndSend(topic, info);
    }

    @Override
    public void run() {
        logger.info("start send message to stomp ...");
            try {
                while(true) {
                    Integer pointId = StorageTaskService.pointIdsQueue.take();
                    ServerResponse indexMsg = realMonitorService.getMonitoringIndexCollectData();
                    ServerResponse monitoringMsg = realMonitorService.getMonitoringCollectData(pointId);
                    this.pushMessage(ConCast.StompDestination.INDEX_LARGE_SCREEN_PATH,
                            JsonUtil.obj2String(indexMsg));
                    this.pushMessage(ConCast.StompDestination.TOPIC_PREFIX + pointId,
                            JsonUtil.obj2String(monitoringMsg));
                }
            } catch (InterruptedException e) { //删除childThread标示
                logger.info("InterruptedException ...");
            }
    }

}
